package com.navinfo.platform.basicdatastatistics.service.impl;


import com.navinfo.platform.basicdatastatistics.dto.BasicDataYearDto;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.common.utils.DateUtils;
import com.navinfo.platform.common.utils.StatisticsDef;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.api.java.function.MapPartitionsFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.sql.Timestamp;
import java.util.*;

public enum BasicDataYearHandleService implements IDataHandleService {

    INS;

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        Dataset<Row> baiscDataset = castValue(obj.get(0));
        Dataset<BasicDataYearDto> saveDataset = baiscDataset.mapPartitions(rowToBasicDataDto(), Encoders.bean(BasicDataYearDto.class));
        String day= DateUtils.getStartEndDay(configMap.get(ICommonService.DAY_VALUE), StatisticsDef.YEAR)._1;
        String columns = configMap.get("hdfs.year.columns");
        List<String> selectColumns = Arrays.asList(columns.split(","));
        BasicDataService.saveData(spark,saveDataset, configMap, "hdfs.vehicle.year.save.tableName", day, selectColumns);
    }


    @SuppressWarnings("unchecked")
    public static MapPartitionsFunction<Row, BasicDataYearDto> rowToBasicDataDto() {
        return itor -> {
            List<BasicDataYearDto> basicDataDtos = new ArrayList<>();
            while(itor.hasNext()){
                Row row = itor.next();
                BasicDataYearDto basicDataYearDto = new BasicDataYearDto();
                basicDataYearDto.setTid(row.getAs("tid"));
                basicDataYearDto.setTrip_mileage(row.getAs("trip_mileage"));
                basicDataYearDto.setTrip_fuel(row.getAs("trip_fuel"));
                basicDataYearDto.setTrip_duration(row.getAs("trip_duration"));
                String crossCities = row.getAs("cross_cities");
                if(StringUtils.isNotEmpty(crossCities)){
                    Set<String> crossCitiesSet = new HashSet<>(Arrays.asList(crossCities.split(",")));
                    basicDataYearDto.setCross_cities(StringUtils.join(crossCitiesSet, ","));
                }
                basicDataYearDto.setRun_days(row.getAs("run_days"));
                basicDataYearDto.setUpdate_time(new Timestamp(System.currentTimeMillis()));
                basicDataDtos.add(basicDataYearDto);
            }
            return basicDataDtos.iterator();
        };
    }


}
